package com.fastmms;

import com.fastmms.analysis.AnalysisBlock;
import com.fastmms.analysis.AnalysisFileUtil;
import com.fastmms.analysis.AnalysisMessage;
import com.fastmms.generate.GenerateMessageBody;
import com.fastmms.generate.GenerateUtil;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;

public class Test {

    public static void main(String[] args) {
        Test.start();
    }

    public static void start() {


        String msgfile = System.getProperty("user.dir") + "/testFile/1w_消息文件.mq";
        /**
         * 测试 生成消息数据
         */
        int bodeSize = 1;
        List<GenerateMessageBody> bodyList = new ArrayList<>(bodeSize);
        int size = 0;
        long msgStartNo = 0;
        for (int i = 0; i < bodeSize; i++) {
            try {
                GenerateMessageBody body = new GenerateMessageBody();
                String subPath = System.getProperty("user.dir") + "/testFile/" + UUID.randomUUID()  +  ".temp";
                body.start(subPath , i,msgStartNo);
                for (int j = 0; j < 1000; j++) {
                    size++;
                    String value = String.valueOf(System.currentTimeMillis() / ((Math.random() * 100000) + 1));
                    body.add(("这是第" + size + "条数据").getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8));
                }
                msgStartNo = body.end();
                bodyList.add(body);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        try {
            GenerateUtil.merge(msgfile, bodyList);
        } catch (Exception e) {
            e.printStackTrace();
        }

        //  解析
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        /**
         * 测试代码块 解析消息数据
         */
        long start = System.currentTimeMillis();
        AtomicLong num = new AtomicLong();
        AnalysisFileUtil.handleFile(msgfile, null,null, new AnalysisMessage() {
            @Override
            public void ready(List<AnalysisBlock> analysisBlockList, Object param) {
               System.out.println("开始执行消息块读取。消息块数量：" + analysisBlockList.size());
               for(AnalysisBlock block : analysisBlockList){
                   System.out.println("开始执行消息块读取。消息块开始地址 ：" + block.getStartPoint() +" / 消息块结束地址：" + block.getEndPoint() );
               }
            }

            @Override
            public void apply(long blockNo, long msgNo,long msgLingNo, byte[] keyData, byte[] valueData, Object param) {
                // 最好是输出到文档中，因为控制台会截断日志 导致无法判断是否正常解析
                String msg =  "消息块序号：" + blockNo + "/ 消息块序号：" + msgNo + "/ 消息行号：" + msgLingNo +  " - key : " + new String(keyData, StandardCharsets.UTF_8) + "  value : " + new String(valueData,StandardCharsets.UTF_8) + "\n";
                System.out.println(msg);
                num.incrementAndGet();
            }

            @Override
            public void finish(long readCount, Object param) {
                System.out.println("文件读取结束，已读取消息： " + readCount + "条，耗时：" + (System.currentTimeMillis() - start) + "毫秒");
            }

            @Override
            public void error(AnalysisBlock errorBlock, long errorPoint, Object param, Exception e) {
                System.out.println("读取消息块失败！读取错误点：" + errorPoint + (errorBlock != null ? ("错误消息块：" + errorBlock) : "")
                        + "，异常消息： " + e.getMessage());
                e.printStackTrace();
            }
        });

        System.out.println("共输出：" + num.get());
    }
}
